package com.mxd.flink.connector.table;

import com.mxd.flink.connector.config.RedisCmd;
import com.mxd.flink.connector.config.RedisOptions;
import com.mxd.flink.connector.operator.RedisOperator;
import com.mxd.flink.connector.operator.RedisOperators;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.table.data.RowData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.TimeUnit;

/**
 * 真正完成按顺序读取redis中数据的函数对象，注意这里只能读取redis中的list结构，因为
 * @author rongdi
 * @date 2022/9/18 13:21
 */
public class RedisScanFunction extends RichSourceFunction<RowData> implements ResultTypeQueryable<RowData> {

    private static final Logger LOG = LoggerFactory.getLogger(RedisScanFunction.class);

    /**
     * flinksql with里的参数列表
     */
    private ReadableConfig config;

    /**
     * redis操作对象
     */
    private RedisOperator redisOperator;

    /**
     * 访问的redis的key值
     */
    private final String key;

    /**
     * 访问的redis的命令
     */
    private final String command;

    /**
     * 每次批量获取redis数据的条数
     */
    private final long batchFetchRows;

    /**
     * 反序列化工具
     */
    private final DeserializationSchema<RowData> deserializer;

    /**
     * 标记当前function的运行状态，volatile修饰让线程之间可见
     */
    private volatile boolean isRunning = true;

    public RedisScanFunction(ReadableConfig config, DeserializationSchema<RowData> deserializer) {
        this.config = config;
        this.key = config.get(RedisOptions.KEY);
        this.command = config.get(RedisOptions.COMMAND);
        this.batchFetchRows = config.get(RedisOptions.BATCH_FETCH_ROWS);
        this.deserializer = deserializer;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return null;
    }

    /**
     * 用来初始化操作redis的工具对象
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        redisOperator = RedisOperators.get(config);
        LOG.info("RedisScanFunction Open");
    }

    /**
     * 执行从redis中读取数据，然后调用sourceContext的collect将数据丢出去
     *
     * @param sourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceContext<RowData> sourceContext) throws Exception {

        /**
         * 这里死循环读取数据
         */
        while(isRunning) {

            Collection valueList = null;
            if(RedisCmd.BLPOP.equalsIgnoreCase(command) || RedisCmd.LPOP.equalsIgnoreCase(command)) {
                valueList = redisOperator.lpop(key,batchFetchRows);
            } else if(RedisCmd.BRPOP.equalsIgnoreCase(command) || RedisCmd.RPOP.equalsIgnoreCase(command)) {
                valueList = redisOperator.rpop(key,batchFetchRows);
            } else if(RedisCmd.SPOP.equalsIgnoreCase(command)) {
                valueList = redisOperator.spop(key,batchFetchRows);
            } else {
                throw new RuntimeException("no match command,such as blpop、lpop、brpop、rpop、spop");
            }
            if(CollectionUtils.isEmpty(valueList)) {
                /**
                 * 这里很大可能是获取数据超时了，很可能是没有那么多数据可以拿，可以多等一下
                  */
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                    /**
                     * 为了调试耗时可以在这里修改isRunning = false表示直到读不到数据了，直接关闭程序,
                     * 这里根据一个是否测试的标识判断，比较合适一点
                     */
                    if(RedisOptions.IS_TEST) {
                        isRunning = false;
                    }

                } catch (InterruptedException e) {
                }
                continue;
            }
            /**
             * 使用反序列化工具反序列化之后循环输出
             */
            for(Object value : valueList) {
                sourceContext.collect(this.deserializer.deserialize(value.toString().getBytes()));
            }

        }
    }

    /**
     * 释放操作redis工具类
     *
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        if (this.redisOperator != null) {
            this.redisOperator.close();
        }
        LOG.info("RedisScanFunction Close");
    }

    @Override
    public void cancel() {
        isRunning = false;
        if (this.redisOperator != null) {
            try {
                this.redisOperator.close();
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
            }
        }
    }


}
